package main

import (
	"os"
	"os/signal"
	"path"
	"syscall"

	"github.com/kpango/glg"
	crono "github.com/utahta/go-cronowriter"
	"gopkg.in/alecthomas/kingpin.v2"
)

var (
	BasePath    = kingpin.Flag("path", "base path to place result data").Default(".").String()
	LogFile     = kingpin.Flag("log", "name of log file").Default("crmlog-consumer.log").Short('l').String()
	EverlogUrl = kingpin.Flag("eurl", "multiple Everylog Urls seperated by ',' (format: 'ip1:port,ip2:port')").Default("127.0.0.1:80").String()
	Interval    = kingpin.Flag("interval", "interval to read kafka, in minute").Default("60").Short('i').Int()
	KUrl        = kingpin.Flag("kurl", "Kafka url (forma: 'ip:port')").Default("127.0.0.1:9092").String()
	KTopic      = kingpin.Flag("ktopic", "Kafka topic").Default("crmlog").String()
	ProPort		= kingpin.Flag("pport", "Prometheus port").Default("10113").String()
)

func main() {
	kingpin.Parse()
	logpath := path.Join(*BasePath, *LogFile)
	logger := crono.MustNew(logpath+".%Y%m%d", crono.WithInit())
	defer logger.Close()
	glg.Get().SetMode(glg.WRITER).AddWriter(logger)
	kh, err := NewKafkaHandler()
	if err != nil {
		glg.Errorf("kafkaConsume.NewKafkaHandler: Failed to Create Kafka consumer: Url:%s, Topic:%s", *KUrl, *KTopic)
		os.Exit(1)
	}
	go kh.matrix.ms.ListenPrometheus(":"+*ProPort)
	go func() {
		signalChan := make(chan os.Signal, 1)
		signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM, syscall.SIGPIPE)
		a := <-signalChan
		glg.Infof("kafkaConsume.main: Received signal %v, close kafka consumer", a)
		if err := kh.Close(); err != nil {
			glg.Infof("kafkaConsume.main: Failed to close kafka: %v", err)
		}
	}()
	kh.Consume()
}
